package com.atguigu.edu.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.util.DateFormatUtil;
import com.atguigu.edu.util.KafkaUtil;
import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.util.Collector;

import java.text.DateFormat;

public class DwdTrafficUniqueVisitorDetail {
    public static void main(String[] args) throws Exception {
        // TODO 1 环境准备
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // TODO 2 设置状态后端

        // TODO 3 读取page主题数据
        String page_topic = "dwd_traffic_page_log";
        String groupId =
                "dwd_traffic_unique_visitor_detail";
        DataStreamSource<String> pageStream =
                env.addSource(KafkaUtil.getFlinkKafkaConsumer(page_topic,
                        groupId));

        // TODO 4 转换和过滤数据 过滤出每个会话的第一条数据 (过滤 last_page_id != null 的数据)
        SingleOutputStreamOperator<JSONObject> jsonObjStream = pageStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                String lastPageId = jsonObject.getJSONObject("page").getString(
                        "last_page_id");
                if (lastPageId == null) {
                    out.collect(jsonObject);
                }
            }
        });

        // TODO 5 根据设备id 分组
        KeyedStream<JSONObject, String> keyedStream =
                jsonObjStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                return value.getJSONObject("common").getString("mid");
            }
        });

        // TODO 6 根据状态日期去重
        SingleOutputStreamOperator<JSONObject> processStream =
                keyedStream.process(new KeyedProcessFunction<String, JSONObject,
                JSONObject>() {
            ValueState<String> lastVisitDtState = null;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> lastVisitDtDescriptor =
                                new ValueStateDescriptor<>("last_visit_dt",
                                        String.class);
                        // 设置存活时间
                        lastVisitDtDescriptor.enableTimeToLive(StateTtlConfig
                                .newBuilder(Time.days(1L))
                                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                                .build());
                        lastVisitDtState =
                                getRuntimeContext().getState(lastVisitDtDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject value, Context ctx,
                                               Collector<JSONObject> out) throws Exception {
                        String lastVisitDt = lastVisitDtState.value();
                        String visitDt = DateFormatUtil.toDate(value.getLong("ts"));
                        if (lastVisitDt == null || !lastVisitDt.equals(visitDt)) {
                            // 每天的第一个数据
                            out.collect(value);
                            lastVisitDtState.update(visitDt);
                        } else {
                            // 状态今天的日期
                        }
                    }
                });

        // TODO 7 过滤出的独立访客数据再写回kafka
        String targetTopic =
                "dwd_traffic_unique_visitor_detail";
        processStream.map(new MapFunction<JSONObject, String>() {
            @Override
            public String map(JSONObject value) throws Exception {
                return value.toJSONString();
            }
        }).addSink(KafkaUtil.getFlinkKafkaProducer(targetTopic));

        // TODO 8 执行任务
        env.execute(groupId);
    }
}
